should we use active instead of leader to stay consistent with streams terminology?

recovering -> restoring? (recovering talks about task state and not store state?)

standardize on queryStaleState vs queryStaleData?

oops. slipped past me.

any reason to swtich to includeStandbyAndRecovering? could we keep queryStaleState?

I had been thinking about it as a directive internally about which stores to bundle, versus an API statement about which consistency level to query at, but upon reflection, even the API is just getting a store, not actually doing a query. I'm standardizing both terms to be the same.

this is creative :)

state.isRunning() is as below

public boolean isRunning() {
            return equals(RUNNING) || equals(STARTING) || equals(PARTITIONS_REVOKED) || equals(PARTITIONS_ASSIGNED);
        }

if its not too much work, we can rename to something like state.isAlive() , that captures what we want to check .. your call.

In the case where includeStandbyAndRecovering=false and the state is RUNNING.. We still want to disallow queries on standbys? from code below, it seems like we will enter the if block and loop over all active and standby tasks?

really nice catch! Thanks.

Regarding isRunning, I had a similar thought, but couldn't remember if StreamThread.State is exposed in the public API. After some code analysis, I don't think it is, so I've renamed it.

To close out the earlier thread.. This test is okay, since NOT_RUNNING will make that instance go to DEAD state (or some non functional state like that) where the store cannot be obtained.. the lines below check that we can stil retrieve the keys from the other replica

is the kafka way to align this with the ( .

It's not fixed in the style guide. The only thing is says is either to put all params on a single line or put each param on a separate line.

What I've seen (which I think makes sense) is, if the line's not too long, and it doesn't hinder readability, you can either do this:

myMethod(one
         two,
         three);

or

myMethod(
  one,
  two,
  three
);

Experience says that either one might be more readable in different circumstances, so it makes sense to allow both. They're both equally obvious regarding where all the params are and where the method call ends. The latter one uses more vertical space for simple method calls, but is more friendly when some of the arguments are themselves method calls.

Thanks for the explanation. I was more curious about this, than anything.. Personally, the latter is very natural and easy to do. I just go with the flow, in these things. :)

Agreed. I also prefer the latter. Thanks for asking! It's way easier to review code that already conforms to the norm, whether or not it's required by the style guide.

tthese verifyAllKVKeys methods will call final ReadOnlyKeyValueStore<String, Long> store = streamsWithKey.store(storeName, QueryableStoreTypes.keyValueStore());, which should fail by default on the standby? (given thats changed from original PR)?

Worth having two tests for both modes?

yeah, that's correct. The test does fail on this version of the code. Nice catch!

The test for active replicas would pass regardless, it's just a question of whether it returns the data during recovery or after the transition to running. But the test you added for standbys would only pass if we allow querying stale state.

I'm not sure it makes sense to add a negative test for standbys, though, since there are already tests ensuring that we throw the appropriate exceptions when using the old method (with a default of no stale stores). WDYT?

makes sense... I will take another look at this when I rebase against the lag PR

cosmetic: extra space at the start

Yeah, I noticed that, too. Oh well.

thinking aloud: guess there is nt much value in wrapping a single provider.. so +1

Yeah, the role of the wrapping store provider really is nothing more than just iterating over the list of providers. For global stores, there's always exactly one provider, so it's purely a performance penalty for no gain at all.

thinking aloud: even though we are binding to a specific StateStoreProvider implementation here, it seems fine,since there are n't any other really in a topology

This is part of the point of this change, actually, StreamThreadStateStoreProvider is no longer a StateStoreProvider. If you need a StateStoreProvider, you have to adapt it with WrappingStoreProvider.

With this change, just Global and Wrapping StateStoreProviders exist? IIUC, most of the xxxStore classes are just accessing the Wrapping.. store provider? Makes me wonder, if we should just use the QueryableStoreProvider everywhere and cull the interface..

Anyway, I am not familiar enough with this part of the code. So I leave it to you..

I also feel that we can collapse the layered interfaces a bit further after we've changed this, since the original motivation of having it is just to "stitch" the global stores and local stores together when exposing as KafkaStreams#stores.

We can consider that in a separate, cleanup PR afterwards.

No, StateStoreProvider is used in the public API as part of QueryableStoreType. Its role is well defined already. That's the one that we should continue to use "everywhere".

QueryableStoreProvider is a utility for selecting either the global store to return OR packaging the thread store providers into a WrappingStoreProvider and using it to actually create a usable store. It's not a StateStoreProvider at all.

WrappingStoreProvider on the other hand is a StateStoreProvider, and adapts a List<StreamThreadStoreProvider> to the StateStoreProvider interface.

GlobalStoreProvider is also a StateStoreProvider.

So, if anything, we could increase clarity by renaming some of these classes to more accurately reflect their roles, but all the involved classes have well defined and necessary roles right now.

@guozhangwang I think we wrote our comments at the same time. Just to respond directly, I do think that we can clean up the code base in a follow-on change (by renaming stuff that's named misleadingly now), but I don't think we can collapse the hierarchy at all.

Thanks! SGTM.

I can work on the follow-on change.

WrappingStoreProvider -> WrappingStateStoreProvider
GlobalStoreProvider -> GlobalStateStoreProvider

QueryableStoreProvider is left as is.

cool. this should now allow standby to be queried

Yep!

I also feel that we can collapse the layered interfaces a bit further after we've changed this, since the original motivation of having it is just to "stitch" the global stores and local stores together when exposing as KafkaStreams#stores.

We can consider that in a separate, cleanup PR afterwards.